package cn.callmee.springboot.pulsar.starter.client.domain;

import cn.callmee.springboot.pulsar.starter.client.exceptions.ClientInitException;
import cn.callmee.springboot.pulsar.starter.client.holder.Holder;
import cn.callmee.springboot.pulsar.starter.client.properties.PulsarProperties;
import com.google.common.base.Strings;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.SubscriptionType;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.stream.Collectors;

public class PulsarUrlGenerator {
    @Resource
    protected PulsarProperties pulsarProperties;

    public String buildTopicUrl(String topic) {
        return pulsarProperties.getConsumer().getDefaultPersistence() + "://" + pulsarProperties.getTenant() + "/" + pulsarProperties.getNamespace() +
                "/" + topic;
    }

    public String buildTopicUrl(String topic, String namespace) {
        if (Strings.isNullOrEmpty(namespace)) {
            return buildTopicUrl(topic);
        }

        return pulsarProperties.getConsumer().getDefaultPersistence() + "://" + pulsarProperties.getTenant() + "/" + namespace + "/" + topic;
    }

    public String buildPulsarConsumerName(String customConsumerName, String generatedConsumerName) {
        if (Strings.isNullOrEmpty(customConsumerName)) {
            return pulsarProperties.getConsumer().getConsumerNamePrefix() + pulsarProperties.getConsumer().getConsumerNameDelimiter() + generatedConsumerName;
        }

        return customConsumerName;
    }

    public String buildPulsarSubscriptionName(String customSubscriptionName, String consumerName) {
        if (Strings.isNullOrEmpty(customSubscriptionName)) {
            return pulsarProperties.getConsumer().getSubscriptionNamePrefix() + pulsarProperties.getConsumer().getConsumerNameDelimiter() + consumerName;
        }

        return customSubscriptionName;
    }

    public SubscriptionType getSubscriptionType(Holder.ConsumerHolder holder) throws ClientInitException {
        return getSubscriptionType(
                Arrays.stream(
                        holder.getAnnotation().subscriptionType()
                ).findFirst().orElse(pulsarProperties.getConsumer().getSubscriptionType()));
    }

    public SubscriptionType getSubscriptionType(SubscriptionType type) throws ClientInitException {
        return type;
    }

    public void buildDeadLetterPolicy(int maxRedeliverCount, String deadLetterTopic, ConsumerBuilder<?> consumerBuilder) {
        DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterBuilder = null;

        if (pulsarProperties.getConsumer().getDeadLetterPolicyMaxRedeliverCount() >= 0) {
            deadLetterBuilder =
                    DeadLetterPolicy.builder().maxRedeliverCount(pulsarProperties.getConsumer().getDeadLetterPolicyMaxRedeliverCount());
        }

        if (maxRedeliverCount >= 0) {
            deadLetterBuilder =
                    DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliverCount);
        }

        if (deadLetterBuilder != null && !deadLetterTopic.isEmpty()) {
            deadLetterBuilder.deadLetterTopic(buildTopicUrl(deadLetterTopic));
        }

        if (deadLetterBuilder != null) {
            consumerBuilder.deadLetterPolicy(deadLetterBuilder.build());
        }
    }

    public String buildConsumerName(Class<?> clazz, Method method) {
        return clazz.getName()
                + pulsarProperties.getConsumer().getConsumerNameDelimiter()
                + method.getName()
                + "_"
                + Arrays
                .stream(method.getGenericParameterTypes())
                .map(Type::getTypeName)
                .collect(Collectors.joining(pulsarProperties.getConsumer().getConsumerParameterDelimiter()));
    }
}
